package com.moon.homework.java;

import org.junit.Test;

import java.util.concurrent.*;

public class MultiThreadSort {

    /**
     * 排序结点
     */
    private static class Node implements Comparable<Node>{
        // 排序层次, level=1时代表区间内排序,level>1时代表两个区间需要合并
        int level;
        // 每层排序的数据区间号
        int id;
        // 数据区间的最小索引
        int low;
        // 数据区间的最大索引
        int high;
        // 待排序数组
        int[] array;
        // 与待排序数组等长的数组
        int[] aux;

        public Node(int level, int id, int low, int high, int[] array, int[] aux) {
            this.level = level;
            this.id = id;
            this.low = low;
            this.high = high;
            this.array = array;
            this.aux = aux;
        }

        @Override
        public int compareTo(Node o) {
            if(this.level != o.level){
                return Integer.compare(this.level, o.level);
            } else {
                return Integer.compare(this.id, o.id);
            }
        }

        @Override
        public String toString() {
            return "Node{" +
                    "level=" + level +
                    ", id=" + id +
                    ", low=" + low +
                    ", high=" + high +
                    '}';
        }
    }

    /**
     * 第一层数据结点排序,区间内排序
     */
    private static class LevelOneSortWorker implements Runnable{
        PriorityBlockingQueue<Node> queue;
        Node node;

        LevelOneSortWorker(Node node, PriorityBlockingQueue<Node> queue) {
            this.node = node;
            this.queue = queue;
        }

        @Override
        public void run() {
            // 区间内排序可以使用任意排序算法
            // SortUtils.mergeSort(node.array, node.low, node.high, node.aux);
            // SortUtils.randomizedQuickSort(node.array, node.low, node.high);
            SortUtils.quickSort(node.array, node.low, node.high);
            // 排序后将node放入queue
            queue.offer(node);
        }
    }

    /**
     * 合并两个有序数据区间的线程
     */
    private static class MergeSortWorker implements Runnable{
        PriorityBlockingQueue<Node> queue;
        Node a;
        Node b;

        MergeSortWorker(Node a, Node b, PriorityBlockingQueue<Node> queue) {
            this.a = a;
            this.b = b;
            this.queue = queue;
        }

        @Override
        public void run() {
            // 将合并后的结点node放入queue
            Node node = mergeNode(a, b);
            queue.offer(node);
            // System.out.println(Thread.currentThread().getName() + " " + node + " 排序后数据区间是否有序: "+SortUtils.isSorted(node.array, node.low, node.high));
        }
    }

    @Test
    public void sortNodeTest() throws InterruptedException {

        PriorityBlockingQueue<Node> priorityQueue = new PriorityBlockingQueue<>();
        priorityQueue.offer(new Node(0, 1, 0, 0, null, null));
        priorityQueue.offer(new Node(2, 2, 0, 0, null, null));
        priorityQueue.offer(new Node(2, 3, 0, 0, null, null));
        priorityQueue.offer(new Node(0, 4, 0, 0, null, null));
        priorityQueue.offer(new Node(3, 5, 0, 0, null, null));
        priorityQueue.offer(new Node(2, 1, 0, 0, null, null));
        priorityQueue.offer(new Node(0, 5, 0, 0, null, null));
        while (!priorityQueue.isEmpty()) {
            // take阻塞式操作, 没有元素时会阻塞.
            System.out.println(priorityQueue.take());
        }
        // priorityQueue.take();
    }

    /**
     * 返回小于输入数n的最大2次幂
     * @param n 输入数n>0
     * @return 不小于输入数n的最大2次幂
     */
    public static int floor2Power(int n){
        // Integer.numberOfLeadingZeros 最高非零位前面的n个0的个数
        return (int) Math.pow(2, 31 - Integer.numberOfLeadingZeros(n));
    }

    @Test
    public void test2Power(){
        for (int i = 0; i < 66; i++) {
            System.out.println(i+" floor: " + floor2Power(i));
            System.out.println(i+"  ceil: " + ceil2Power(i));
        }
        System.out.println(ceil2Power(Integer.MAX_VALUE));
        System.out.println(floor2Power(Integer.MAX_VALUE));
    }

    /**
     * 不小于一个数的最小2的幂次方, 对于10就是16,对于21就是 32
     * @param c 输入值, 一个大于0的数
     * @return 不小于cores的最小2的幂次方
     */
    private static int ceil2Power(int c) {
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        //一个类型int数据共有32位, 上述过程就是使n的二进制数的低位全部变为1
        return (n < 0) ? 1 : n + 1;
    }

    // 获取cpu个数
    private final static int CPU_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 多线程排序
     * @param array 待排序数组
     */
    public static void multiThreadSort(int[] array) {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(CPU_COUNT, CPU_COUNT,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        PriorityBlockingQueue<Node> priorityBlockQueue = new PriorityBlockingQueue<>();

        int length = array.length;
        // 数据区间数
        int segment;

        // 初始设置每个数据区间的数据个数为100万
        int segmentNumberCount = 1000000;
        if (length < segmentNumberCount) {
           // 数组长度小于100万不进行分区
           segment = 1;
        } else {
            // 先按每个数据区间100万个数据来对原数组进行划分
            if (length % segmentNumberCount == 0) {
                segment = length / segmentNumberCount;
            } else {
                segment = length / segmentNumberCount + 1;
            }

            // 最终segment取CPU核数和segment较大的值
            segment = Math.max(segment, CPU_COUNT);
            // 取小于分区数的最大的2的次幂, 保证分区块是2的次幂. 比如输入10返回16
            segment = ceil2Power(segment);
            // segment = floor2Power(CPU_COUNT);
        }

        // 最终根据分区数设置区间内的数据个数
        segmentNumberCount = length/segment;
        System.out.println("数据分区数: " + segment);

        // 与原始数组同长度的额外数据, 两个数据区间的排序用.
        int[] aux = new int[length];
        // 数据区间编号
        int id = 1;
        int beginIndex = 0;
        // 前segment-1个区间加入到线程池排序
        for (int i = 0; i < segment-1; i++) {
            // 构造第一层的排序结点node
            Node node = new Node(1, id++, beginIndex,beginIndex + segmentNumberCount - 1, array, aux);
            executor.execute(new LevelOneSortWorker(node, priorityBlockQueue));
            beginIndex += segmentNumberCount;
        }

        // 最后一个数据进入线程池排序, 因为最后一个区间可能会大于设定的数据长度
        Node node = new Node(1, id, beginIndex,length - 1, array, aux);
        executor.execute(new LevelOneSortWorker(node, priorityBlockQueue));

        while (true) {
            try {
                Node a = priorityBlockQueue.take();
                // 判断数组排序是否结束
                if (a.low == 0 && a.high == length - 1) {
                    // 排序结束销毁线程池
                    executor.shutdown();
                    break;
                }
                Node b = priorityBlockQueue.take();

                // 同层之间排序
                // id为1和2进行合并, 3和4合并, 5和6合并...
                if (canMerge(a, b)) {
                    executor.submit(new MergeSortWorker(a, b, priorityBlockQueue));
                } else {
                    // 这里再取一个进行判断, 避免总是取到同一对无法合并的a和b
                    Node c = priorityBlockQueue.take();
                    if (canMerge(a, c)) {
                        executor.submit(new MergeSortWorker(a, c, priorityBlockQueue));
                        priorityBlockQueue.offer(b);
                    } else if (canMerge(b, c)) {
                        executor.submit(new MergeSortWorker(b, c, priorityBlockQueue));
                        priorityBlockQueue.offer(a);
                    } else {
                        priorityBlockQueue.offer(a);
                        priorityBlockQueue.offer(b);
                        priorityBlockQueue.offer(c);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 判断数据区间结点a和b结点是否能合并
     * 可以合并的条件:
     *       a.id为奇数, a.id + 1 = b.id
     *     或 a.id为偶数 b.id + 1 = a.id
     * @param a 数据区间结点a
     * @param b 数据区间结点b
     * @return true|false
     */
    private static boolean canMerge(Node a, Node b){
        if(a.level == b.level){
            if (a.id % 2 != 0 && a.id + 1 == b.id) {
                // a.id为奇数 a.id + 1 = b.id
                return true;
            } else if (a.id % 2 == 0 && b.id + 1 == a.id) {
                // a.id为偶数 b.id + 1 = a.id
                return true;
            }
        }
        return false;
    }


    /**
     * 利用归并排序的思想, 合并同一层相邻的两个node中的有序数组
     * 同层相邻: a.level = b.level
     * a.id为偶数则 a.id+1 = b.id
     * a.id为奇数则 a.id - 1 = b.id
     * @param a 数据区间结点1
     * @param b 数据区间结点2
     */
    private static Node mergeNode(Node a, Node b){
        int[] array = a.array;
        int[] aux = a.aux;
        int level = a.level + 1;
        int low, mid, high, id;
        if (a.id < b.id) {
            low = a.low;
            mid = a.high;
            high = b.high;
            id = b.id/2;
        } else {
            low = b.low;
            mid = b.high;
            high = a.high;
            id = a.id/2;
        }
        // 归并排序合并两个有序数据
        SortUtils.mergeArray(array, low, mid, high, aux);
        Node node = new Node(level, id, low, high, array, aux);
        // System.out.println(Thread.currentThread().getName() + " " + node + " 排序后数据区间是否有序: "+SortUtils.isSorted(node.array, node.low, node.high));
        return node;
    }
}
